package com.rem.flink.flink10Sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 使用 开窗聚合 OVER 普通TopN 取值
 *
 * @author Rem
 * @date 2022-11-08
 */

public class OverTopNTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 在创建表的DDL中直接定义时间属性
        String createDdl = "CREATE TABLE clickTable (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts BIGINT, " +
                " et AS TO_TIMESTAMP( FROM_UNIXTIME(ts / 1000) ), " +  //BIGINT类型转为时间戳类型
                " WATERMARK FOR et AS et - INTERVAL '1' SECOND " +    //定义1s延迟时间的水位线
                ") WITH (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'input/clicks.csv', " +
                " 'format' =  'csv' " +
                ")";

        tableEnv.executeSql(createDdl);

        /**
         * 选取当前用户中浏览量最大的2个
         */
        Table topNtable = tableEnv.sqlQuery("select user, cnt, row_num " +
                "from (" +
                " select * ,ROW_NUMBER() OVER (order by cnt desc) as row_num " +
                "from (select user, count(url) as cnt from clickTable group by user) " +
                ") where row_num <=2"
        );

        tableEnv.toChangelogStream(topNtable).print();

        env.execute();
    }
}
